package com.allenjsl.springactuator.service;


import com.alibaba.fastjson.JSON;
import com.allenjsl.springactuator.config.BeanContextAware;
import com.allenjsl.springactuator.entity.User;
import com.allenjsl.springactuator.rocketmq.OrderMessageQueueSelector;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.UnsupportedEncodingException;
import java.util.UUID;

/**
 * @ClassName ProducerService
 * @Description 生产者服务类
 * @Author panchen
 * @CreateDate 2021/4/19 2:18 下午
 * @Version V1.0
 */
@Slf4j
@Service
public class ProducerService {
    @Value("${rocketmq.producer.send-message-timeout}")
    private Integer messageTimeOut;

    // 建议正常规模项目统一用一个TOPIC
    @Value("${rocketmq.producer.topic}")
    private  String topic;


    // 直接注入使用，用于发送消息到broker服务器
//    @Autowired
//    private RocketMQTemplate rocketMQTemplate;

    private static final String Tag = "charge";


    private static final String[] ORDER_MESSAGES = {"下单","结算","支付","完成"};

    /**
     * 先向MQ Server发送半消息
     * @param user 用户信息
     */
    public void sendHalfMsg(User user) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        // 生成生产事务id
        String transactionId = UUID.randomUUID().toString().replace("-", "");
        log.info("【发送消息】transactionId={}", transactionId);
        log.info(JSON.toJSONString(user));

        OrderMessageQueueSelector orderMessageQueueSelector = new OrderMessageQueueSelector();
        DefaultMQProducer producer = BeanContextAware.getBean("testProducer");
        //注：要实现顺序消费，必须同步发送消息

        for (int i = 0;i < 3;i++) {
            String orderId = String.valueOf(System.currentTimeMillis()) + (i + 1);
            for (int j = 0, size = ORDER_MESSAGES.length; j < size; j++) {
                String message = "Order-" + orderId + "-" + ORDER_MESSAGES[j];
                String keys = message;
                byte[] messageBody = new byte[0];
                try {
                    messageBody = message.getBytes("UTF-8");
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
                Message mqMsg = new Message("PM_TEST_TOPIC", Tag, keys, messageBody);
                producer.send(mqMsg, orderMessageQueueSelector, i);
            }
        }
        //发送自定义对象
//        rocketMQTemplate.convertAndSend("PM_TEST_TOPIC" + ":" + "tag1",user);//会自动转换为json格式
//        // 发送事务消息（参1：生产者所在事务组，参2：topic+tag，参3：消息体(可以传参)，参4：发送参数）
//        TransactionSendResult sendResult = rocketMQTemplate.sendMessageInTransaction(topic + ":" + "tag1",
//                MessageBuilder.withPayload(user).setHeader(RocketMQHeaders.TRANSACTION_ID, transactionId).build(),
//                user);
//        log.info("【发送消息】sendResult={}", JSON.toJSONString(sendResult));
//        return sendResult;
    }


//    /**
//     * 普通发送（这里的参数对象User可以随意定义，可以发送个对象，也可以是字符串等）
//     */
//    public void send(User user) {
//        rocketMQTemplate.convertAndSend(topic + ":tag1", user);
////        rocketMQTemplate.send(topic + ":tag1", MessageBuilder.withPayload(user).build()); // 等价于上面一行
//    }

//    /**
//     * 发送同步消息（sendResult为返回的发送结果）
//     */
//    public SendResult sendMsg(String msgBody) {
//        SendResult sendResult = rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(msgBody).build());
//        log.info("【sendMsg】sendResult={}", JSON.toJSONString(sendResult));
//        return sendResult;
//    }

//    /**
//     * 发送异步消息（适合对响应时间敏感的业务场景）
//     */
//    public void sendAsyncMsg(String msgBody) {
//        rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(msgBody).build(), new SendCallback() {
//            @Override
//            public void onSuccess(SendResult sendResult) {
//                // 处理消息发送成功逻辑
//                log.info("发送成功");
//            }
//
//            @Override
//            public void onException(Throwable throwable) {
//                // 处理消息发送异常逻辑
//                log.error("发送失败");
//            }
//        });
//    }

//    /**
//     * 发送延时消息<br/>
//     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
//     */
//    public void sendDelayMsg(String msgBody, int delayLevel) {
//        rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(msgBody).build(), messageTimeOut, delayLevel);
//    }
//
//    /**
//     * 发送单向消息（不关心发送结果，如日志）
//     */
//    public void sendOneWayMsg(String msgBody) {
//        rocketMQTemplate.sendOneWay(topic, MessageBuilder.withPayload(msgBody).build());
//    }
//
//    /**
//     * 发送带tag的消息，直接在topic后面加上":tag"
//     */
//    public SendResult sendTagMsg(String msgBody) {
//        return rocketMQTemplate.syncSend(topic + ":tag2", MessageBuilder.withPayload(msgBody).build());
//    }
}

